package com.alex.projects.bitcask.core;

import com.alex.projects.bitcask.Tools;
import com.alex.projects.bitcask.api.KeyIndexEntry;
import com.alex.projects.bitcask.api.StoreKey;
import com.alex.projects.bitcask.api.StoreValue;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.zip.CRC32;

import static com.alex.projects.bitcask.api.StoreValue.*;

/**
 * 存储服务，提供Key/value存储、查询功能
 */
public class StoreService {

    public static final String FILE_PREFIX = "bitcask_";

    private static final long FILE_SIZE_MAX = 50 * 1024 * 1024 * 8L;

    private static final String DELETE_CONTENT = "$<<DELETE_FLAG>>";

    private static final byte[] DELETE_CONTENT_BYTES;

    private static final int DELETE_CONTENT_SIZE;

    static {
        DELETE_CONTENT_BYTES = DELETE_CONTENT.getBytes(StandardCharsets.UTF_8);
        DELETE_CONTENT_SIZE = DELETE_CONTENT_BYTES.length;
    }

    private String rootDirPath;

    private File rootDir;

    private KeyIndexService keyIndexService;

    private AtomicInteger fileIndex;

    private FileChannel fileChannel;

    private long filePos;

    public StoreService(String rootDirPath) {
        this.rootDirPath = rootDirPath;
        this.keyIndexService = new KeyIndexService();
    }

    public void init() throws IOException {
        checkRootDir();
        lockRootDir();
        if (!recovery()) {
            fileIndex = new AtomicInteger(0);
            openNewFile(getFileName(fileIndex.get()));
        }
    }

    private void lockRootDir() throws IOException {

    }

    private void checkRootDir() throws IOException {
        rootDir = new File(rootDirPath);
        if (!rootDir.exists()) {
            boolean mkdirRet = rootDir.mkdirs();
            if (!mkdirRet) {
                throw new RuntimeException("create root directory failed");
            }
        }
    }

    /**
     * 合并老的数据存储文件
     */
    private void merge() throws IOException {
//        int presentFileIndex = fileIndex.get();
//        int mergeMaxFileIndex =
    }

    /**
     * 恢复索引
     *
     * @return 当不存在旧文件时，返回false；否则返回true
     * @throws IOException 操作旧文件异常
     */
    private boolean recovery() throws IOException {
        File[] childFiles = rootDir.listFiles();
        if (Objects.isNull(childFiles)) {
            return false;
        }
        List<File> oldStoreFileList = new ArrayList<>();
        for (File childFile : childFiles) {
            if (childFile.isDirectory()) {
                continue;
            }
            if (childFile.getName().startsWith(FILE_PREFIX)) {
                oldStoreFileList.add(childFile);
            }
        }
        if (oldStoreFileList.isEmpty()) {
            return false;
        }
        Collections.sort(oldStoreFileList, (left, right) -> {
            String leftFileName = left.getName();
            Integer leftFileIndex = Tools.getFileIndexByName(leftFileName);
            String rightFileName = left.getName();
            Integer rightFileIndex = Tools.getFileIndexByName(rightFileName);
            return leftFileIndex.compareTo(rightFileIndex);
        });
        FileChannel channel;
        int headLen = CRC_LEN + STSAMP_LEN + KEY_SIZE_LEN + VALUE_SIZE_LEN;
        for (int i = 0; i < oldStoreFileList.size(); i++) {
            long begin = 0;
            File oldStoreFile = oldStoreFileList.get(i);
            channel = new RandomAccessFile(oldStoreFile, "rw").getChannel();
            ByteBuffer headerBuf = ByteBuffer.allocate(headLen);
            for (readAndFlip(channel, headerBuf); headerBuf.hasRemaining(); ) {
                StoreValue header = header(headerBuf);
                short keySize = Tools.bytesToShort(header.getKSz());
                int valueSize = Tools.bytesToInt(header.getValueSz());
                ByteBuffer keyBuf = ByteBuffer.allocate(keySize);
                channel.read(keyBuf);
                StoreKey key = new StoreKey(new String(keyBuf.array(), StandardCharsets.UTF_8));
                KeyIndexEntry indexEntry = new KeyIndexEntry();
                indexEntry.setFileId(Tools.getFileIndexByName(oldStoreFile.getName()));
                indexEntry.setTstamp(Tools.bytesToLong(header.getTstamp()));
                indexEntry.setValuePos(begin);
                indexEntry.setValueSz(headLen + keySize + valueSize);
                keyIndexService.putIndex(key, indexEntry);
                begin += indexEntry.getValueSz();
                channel.position(begin);
                headerBuf.clear();
                readAndFlip(channel, headerBuf);
            }
            if (i == oldStoreFileList.size() - 1) {
                int oldMaxFileIndex = Tools.getFileIndexByName(oldStoreFile.getName());
                if (channel.size() < FILE_SIZE_MAX) {
                    fileIndex.set(oldMaxFileIndex);
                    fileChannel = channel;
                    filePos = channel.size();
                } else {
                    fileIndex.set(oldMaxFileIndex + 1);
                    openNewFile(getFileName(fileIndex.get()));
                }
            }
        }
        return true;
    }

    public void readAndFlip(FileChannel channel, ByteBuffer buffer) throws IOException {
        channel.read(buffer);
        buffer.flip();
    }

    public void put(String keyStr, byte[] value) throws IOException {
        StoreKey key = new StoreKey(keyStr);
        long tstamp = System.currentTimeMillis();
        StoreValue storeValue = constructStoreValue(key, value, tstamp);
        int valueLen = storeValue.size();
        if (isFileFull(valueLen)) {
            //文件空间不够，扩展文件
            int fileIndex = this.fileIndex.addAndGet(1);
            String newActiveFileName = getFileName(fileIndex);
            fileChannel.close();
            openNewFile(newActiveFileName);
        }
        //文件空间足够，直接写入
        fileChannel.write(storeValue.wrap());
        //更新索引
        updateIndex(key, tstamp, valueLen);
        //更新文件偏移量
        filePos += valueLen;
    }

    private void updateIndex(StoreKey key, long tstamp, int valueLogSz) {
        //增加索引
        KeyIndexEntry entry = new KeyIndexEntry();
        entry.setFileId(fileIndex.get());
        entry.setTstamp(tstamp);
        entry.setValuePos(filePos);
        entry.setValueSz(valueLogSz);
        keyIndexService.putIndex(key, entry);
    }

    private StoreValue constructStoreValue(StoreKey key, byte[] value, long tstamp) {
        int valueSz = value.length;
        long crc = crc32(value, valueSz);
        StoreValue storeValue = new StoreValue();
        storeValue.setCrc(Tools.longToBytes(crc));
        storeValue.setTstamp(Tools.longToBytes(tstamp));
        storeValue.setKSz(Tools.shortToBytes(key.getKeyLen()));
        storeValue.setValueSz(Tools.intToBytes(valueSz));
        storeValue.setKey(key.getBytes());
        storeValue.setValue(value);
        return storeValue;
    }

    private long crc32(byte[] value, int valueSz) {
        CRC32 crc32 = new CRC32();
        crc32.update(value, 0, valueSz);
        return crc32.getValue();
    }

    private boolean isFileFull(long valueLen) {
        return (filePos + valueLen) >= FILE_SIZE_MAX;
    }

    /**
     * 创建新文件，并将文件指针指向新文件，filePos=0
     *
     * @param newActiveFileName
     * @throws IOException
     */
    private void openNewFile(String newActiveFileName) throws IOException {
        File newFile = new File(newActiveFileName);
        if (!newFile.exists()) {
            boolean createRes = newFile.createNewFile();
            if (!createRes) {
                throw new RuntimeException(String.format("create new log file failed %s", newActiveFileName));
            }
        }
        fileChannel = new RandomAccessFile(newActiveFileName, "rw").getChannel();
        filePos = 0;
    }

    private String getFileName(int fileIndex) {
        return FILE_PREFIX + fileIndex;
    }

    public byte[] get(String keyStr) throws IOException {
        StoreKey key = new StoreKey(keyStr);
        KeyIndexEntry indexEntry = keyIndexService.getIndex(key);
        if (Objects.isNull(indexEntry)) {
            //不存在
            return null;
        }
        int fileId = indexEntry.getFileId();
        ByteBuffer buf = ByteBuffer.allocate(indexEntry.getValueSz());
        if (fileIndex.get() != fileId) {
            FileChannel fileChannel = new RandomAccessFile(getFileName(fileId), "rw").getChannel();
            fileChannel.position(indexEntry.getValuePos());
            fileChannel.read(buf);
            fileChannel.close();
        } else {
            long oldPosition = fileChannel.position();
            fileChannel.position(indexEntry.getValuePos());
            fileChannel.read(buf);
            fileChannel.position(oldPosition);
        }
        StoreValue storeValue = StoreValue.getInstance(buf);
        byte[] valueSzBytes = storeValue.getValueSz();
        int valueSz = Tools.bytesToInt(valueSzBytes);
        if (valueSz == DELETE_CONTENT_SIZE) {
            //已删除
            if (Tools.byteArrEquals(storeValue.getValue(), DELETE_CONTENT_BYTES)) {
                return null;
            }
        }
        return storeValue.getValue();
    }

    public void delete(String keyStr) throws IOException {
        put(keyStr, DELETE_CONTENT.getBytes(StandardCharsets.UTF_8));
    }
}
